package org.niit.service

import org.apache.spark.ml.recommendation.ALSModel
import org.apache.spark.sql.{DataFrame, Dataset, SaveMode}
import org.apache.spark.streaming.dstream.DStream
import org.niit.bean.Answer
import org.niit.common.TService
import org.niit.util.SparkUtil

/**
 * Date:2025/6/12
 * Author：Ys
 * Description:
 */
class EDURecomendService{


   def dataAnalysis(data: DStream[Answer]): Unit = {

     val spark = SparkUtil.takeSpark()
     import spark.implicits._
     import org.apache.spark.sql.functions._

    data.foreachRDD(rdd=>{
      //1.声明模型路径
      val path = "output/als_model/1749699376645"
      //2.加载模型
      val model = ALSModel.load(path)
      //3.由于训练模型的时候不能有任务的中文字样，所在在训练的时候我们就已经对id进行了切割
      //那么在预测的时候，我们也需要对id进行切割，才可以完成预测，所在要定义一个UDF方法

      val id2Int = udf((student_id: String) => {
        student_id.split("_")(1).toInt
      })

      //4.由于ALS模型传入的参数必须是DataFrame，所以要将RDD转换DataFrame
      val answerDF: DataFrame = rdd.toDF
      answerDF.show()

      //5.由于是针对学生id进行推荐错题，所以先将studentID查询并切割出来
      val studentIdDF: DataFrame = answerDF.select(id2Int($"student_id") as "student_id")

      //6.使用模型给学生推荐题目    参数1：学生ID  参数2：推荐题目数量
      //recommendForUserSubset的含义：给用户推荐子集
      val recommendDF: DataFrame = model.recommendForUserSubset(studentIdDF, 10)

      //将推荐的内容 数据结构进行打印
      recommendDF.printSchema()
      recommendDF.show(false)

      //7.处理推荐结果，取出学生id和推荐的题目id,将题目拼接成一个字符串 [id1,id2...]
      val res = recommendDF.as[(Int, Array[(Int, Float)])].map(t => {
        val studentId = "学生ID_" + t._1
        val questionIds = t._2.map(t => "题目ID_" + t._1).mkString(",")
        (studentId, questionIds)
      }).toDF("student_id","recommendations")

      //8.将answerDF 和 res 进行join 合并
      val allInfoDF: DataFrame = answerDF.join(res, "student_id")

      //9.写入数据库
      allInfoDF.write.format("jdbc")
        .option("url", "jdbc:mysql://node1:3306/BD2_2025?useUnicode=true&characterEncoding=utf8")
        .option("driver", "com.mysql.jdbc.Driver")
        .option("user","root")
        .option("password","Niit@123")
        .option("dbtable","edu")
        .mode(SaveMode.Append)
        .save()



    })


  }

}
